package task2.util

import java.io.File

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.internal.Logging

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object FilePathHelper extends Logging {

  val conf = new Configuration()

  /**
   * 检测源路径是否存在，并且创建目标路径
   *
   * @param src    文件源路径
   * @param target 文件目标路径
   */
  def checkSrcAndCreateTarget(src: String, target: String): Unit = {
    val fs = FileSystem.get(conf)
    require(fs.exists(new Path(src)), s"$src 不存在！")
    // 如果 target 路径不存在则创建
    val targetPath = new Path(target)
    if (!fs.exists(targetPath)) fs.mkdirs(targetPath)
  }

  /**
   * 递归扫描源路径所有文件，并构建 源路径 -> 目标路径 的列表
   *
   * @param src 文件源路径
   * @param target 文件目标路径
   * @return
   */
  def buildSrcToTargetList(src: String, target: String): mutable.Seq[(String, String)] = {
    val fs = FileSystem.get(conf)
    logInfo("构建拷贝文件的列表...")
    val files = fs.listFiles(new Path(src), true)
    val copyList = ArrayBuffer[(String, String)]()
    while (files.hasNext) {
      val filePath = files.next().getPath.toString
      val start = filePath.lastIndexOf(src)
      // 截取 source 目录下的所有文件路径，拼接成 target/xxx 的形式并放入数组中
      val file = filePath.substring(start + src.length + 1)
      copyList += ((filePath, s"$target${File.separator}$file"))
    }
    copyList
  }


}
